package com.baixiaowen.disruptor.HighMulti多生产者多消费者模式;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        // 1. 创建RingBuffer
        RingBuffer<Order> ringBuffer = RingBuffer.create(
                ProducerType.MULTI,
                Order::new,
                1024 * 1024,
                new YieldingWaitStrategy()
        );

        // 2. 通过 RingBuffer 创建一个序号屏障
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        // 3. 创建多个消费者数组
        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("C" + i);
        }

        // 4. 构建多消费者工作池
        WorkerPool<Order> workerPool = new WorkerPool<Order>(
                ringBuffer,
                sequenceBarrier,
                new EventExceptionHandler(),
                consumers
        );

        // 5. 设置多个消费者的sequence序号，用于单独统计消费进度，并且设置到ringBuffer中
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

        // 6. 启动WorkerPool
//        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
        workerPool.start(Executors.newFixedThreadPool(5));

        CountDownLatch latch = new CountDownLatch(1);

        for (int i = 0; i < 100; i++) {
            Producer producer = new Producer(ringBuffer);
            new Thread(() -> {
                try {

                } catch (Exception e) {
                    e.printStackTrace();
                }
                for (int j = 0; j < 100; j++) {
                    producer.onData(UUID.randomUUID().toString());
                }
            }).start();
        }

        Thread.sleep(2000);
        System.err.println("-------线程创建完毕，开始生产数据-------");
        latch.countDown();

        Thread.sleep(5000);
        System.err.println("消费总数：" + consumers[2].getCount());
    }

    static class EventExceptionHandler implements ExceptionHandler {
        /**
         * 进行消费的时候的异常处理
         * @param ex
         * @param sequence
         * @param event
         */
        @Override
        public void handleEventException(Throwable ex, long sequence, Object event) {

        }

        /**
         * 启动的时候的异常处理
         * @param ex
         */
        @Override
        public void handleOnStartException(Throwable ex) {

        }

        /**
         * 结束的时候的异常处理
         * @param ex
         */
        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
